package cn.me.alphamq.broker.model.timewheel;

import cn.me.alphamq.broker.core.timewheel.TimeWheelModelExecutor;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 时间轮的模型
 * <p>
 * 当前仅支持 SECONDS MINUTES 的时间单位
 *
 * @author f
 */
public class TimeWheelModel {

    /**
     * 时间的指针
     */
    private volatile int hand;

    /**
     * 时间轮内部的槽位链表的数组
     */
    private List<TimeWheelSlotModel>[] slotListArray;

    /**
     * 时间轮内部每个槽位链表的锁的数组
     */
    private Object[] slotListLockArray;

    /**
     * 比当前时间轮时间单位小一级的时间轮
     */
    private final TimeWheelModel next;

    /**
     * 当前时间轮一个时间单位有多少秒
     */
    private final int secPerTimeUnit;

    /**
     * 处理到时间的消息，清除处理完毕的消息，并且让指针移动一位
     */
    public void exec() {
        int hand = incAndGetHand();
        List<TimeWheelSlotModel> slots = slotListArray[hand];
        synchronized (slotListLockArray[hand]) {
            slots.forEach(slot -> {
                // 如果有剩余时间，则将消息交给下一个时间轮来处理
                int remainTime = slot.getRemainTime();
                if (remainTime != 0) {
                    fireNextWheel(slot.getMsg(), slot.getMsgType(), slot.getExecuteTime(), remainTime);
                    return;
                }

                // 处理消息
                TimeWheelModelExecutor.execute(slot, true);
            });
            slots.clear();
        }
    }

    // 将消息交给下一个时间轮来处理
    private void fireNextWheel(Object msg, int msgType, long executeTime, int remainTime) {
        // 防止出现比 1 秒还小的延迟时间
        if (next == null) {
            throw new RuntimeException("there has no time unit smaller than second!");
        }

        // 将这个消息交给下一个时间轮处理
        int delay = remainTime / next.secPerTimeUnit;
        remainTime = remainTime % next.secPerTimeUnit;
        next.add(delay, msg, msgType, executeTime, remainTime);
    }

    // 让指针移动一位，并获取指针移动后的值
    private int incAndGetHand() {
        synchronized (this) {
            hand = (hand + 1) % slotListArray.length;
        }
        return hand;
    }

    /**
     * 添加一个延迟消息，没有剩余时间
     *
     * @param delay       延迟的时间，范围：[0, {@code slotListArray.length}]
     * @param msg         消息
     * @param msgType     消息类型
     * @param executeTime 处理时间
     */
    public void add(int delay, Object msg, int msgType, long executeTime) {
        add(delay, msg, msgType, executeTime, 0);
    }

    /**
     * 添加一个延迟消息
     *
     * @param delay       延迟的时间，范围：[0, {@code slotListArray.length}]
     * @param msg         消息
     * @param msgType     消息类型
     * @param executeTime 处理时间
     * @param remainTime  剩余时间
     */
    public void add(int delay, Object msg, int msgType, long executeTime, int remainTime) {
        // 如果没有延迟时间，则将其交给下一个时间轮处理
        if (delay == 0) {
            fireNextWheel(msg, msgType, executeTime, remainTime);
            return;
        }

        // 获取当前指针
        int hand;
        synchronized (this) {
            hand = this.hand;
        }

        // 获取消息应该存放的 槽位链表 的下标
        int remain = slotListArray.length - hand - 1;
        int diff = delay - remain;
        int index = diff < 0 ? hand + delay : diff;

        // 将消息存放到指定的 槽位链表 中
        List<TimeWheelSlotModel> slots = slotListArray[index];
        TimeWheelSlotModel slot = new TimeWheelSlotModel(msg, msgType, executeTime, remainTime);
        synchronized (slotListLockArray[index]) {
            slots.add(slot);
        }
    }

    public TimeWheelModel(TimeUnit timeUnit) {
        this(timeUnit, null);
    }

    public TimeWheelModel(TimeUnit timeUnit, TimeWheelModel next) {
        this.next = next;
        if (TimeUnit.SECONDS == timeUnit) {
            init(60);
            this.secPerTimeUnit = 1;
        } else if (TimeUnit.MINUTES == timeUnit) {
            init(60);
            this.secPerTimeUnit = 60;
        } else {
            throw new RuntimeException("not supported time unit type: " + timeUnit);
        }
    }

    // 根据传入的大小初始化每个槽位链表和它对应的锁
    private void init(int size) {
        this.slotListArray = new List[size];
        this.slotListLockArray = new Object[size];
        for (int i = 0; i < size; i++) {
            this.slotListArray[i] = new LinkedList<>();
            this.slotListLockArray[i] = new Object();
        }
    }
}
